package com.lianjia.sh.se.message.event.client.publisher.kafka;

import com.lianjia.sh.se.kafka.client.KafkaProducerFactory;
import com.lianjia.sh.se.message.event.client.util.TopicUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;

/**
 * Created by yangsp1 on 16/8/3.
 */
@Component
public class MessageEventKafkaSenderFactory {
    @Autowired
    KafkaProducerFactory kafkaProducerFactory;

    private final Map<String,MessageEventKafkaSender> senderCache = new HashMap<>();

    /**
     * 创建kafkaSender
     * @param projectCode
     * @param serviceCode
     * @return
     */
    public MessageEventKafkaSender createKafkaSender(String projectCode,String serviceCode){
        String topic = TopicUtil.populateTopic(projectCode,serviceCode);
        MessageEventKafkaSender messageEventKafkaSender = senderCache.get(topic);
        if(messageEventKafkaSender != null){
            return messageEventKafkaSender;
        }
        synchronized (this){
            if(messageEventKafkaSender != null){
                return messageEventKafkaSender;
            }

            messageEventKafkaSender = new MessageEventKafkaSender();
            messageEventKafkaSender.setKafkaProducerFactory(kafkaProducerFactory);
            messageEventKafkaSender.setTopic(topic);
        }
        return messageEventKafkaSender;
    }
}
